package com.github.hongkaiwen.rxjava.play.caselib;

import io.reactivex.Observable;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.schedulers.Schedulers;
import java.util.concurrent.TimeUnit;

/**
 * 线程切换
 * 下一话题： 并行执行
 */
public class RxThreadPlay {

    public static void main(String... args) throws InterruptedException {
        //如果不做特殊指定，被监听者执行线程与发起方线程一样，监听者执行线程与被监听者一样
        Observable.create((ObservableOnSubscribe) emitter -> {
            System.out.println("case1: " + Thread.currentThread());
            emitter.onNext("hello");
        }).subscribe(o -> {
            System.out.println("case1: " + Thread.currentThread());
        });

        //subscribeOn 指定被监听者执行的线程
        Observable.create((ObservableOnSubscribe) emitter -> {
            System.out.println("case2: " + Thread.currentThread());
            emitter.onNext("hello");
        }).subscribeOn(Schedulers.computation()).subscribe(o -> {
            System.out.println("case2: " + Thread.currentThread());
        });


        //observeOn 指定监听者执行的线程
        Observable.create((ObservableOnSubscribe) emitter -> {
            System.out.println("case3: " + Thread.currentThread());
            emitter.onNext("hello");
        }).observeOn(Schedulers.computation()).subscribe(o -> {
            System.out.println("case3: " + Thread.currentThread());
        });

        //如果不做特殊指定，被监听者执行线程与发起方线程一样，监听者执行线程与被监听者一样
        Thread t = new Thread(() -> Observable.create((ObservableOnSubscribe) emitter -> {
            System.out.println("case4: " + Thread.currentThread());
            emitter.onNext("hello");
        }).observeOn(Schedulers.computation()).subscribe(o -> {
            System.out.println("case4: " + Thread.currentThread());
        }));
        t.setName("test");
        t.getThreadGroup();
        t.start();

        //subscribeOn 多次切换，以第一次为准
        Observable.create((ObservableOnSubscribe) emitter -> {
            System.out.println("case5: " + Thread.currentThread());
            emitter.onNext("hello");
        }).subscribeOn(Schedulers.io()).subscribeOn(Schedulers.newThread()).subscribe(o -> {
            System.out.println("case5: " + Thread.currentThread());
        });

        //observeOn 多次切换，以最后一次为准，可以多次切换
        Observable.create((ObservableOnSubscribe) emitter -> {
            System.out.println("case6: " + Thread.currentThread());
            emitter.onNext("hello");
        }).observeOn(Schedulers.newThread()).map(b -> {
            System.out.println("case6: " + Thread.currentThread());
            return b;
        }).observeOn(Schedulers.io()).subscribe(o -> {
            System.out.println("case6: " + Thread.currentThread());
        });


        TimeUnit.SECONDS.sleep(1);
    }
}
